package com.liuyjy.es.service;

import com.alibaba.fastjson2.JSON;
import com.liuyjy.core.entity.PageResult;
import com.liuyjy.core.exception.EsException;
import com.liuyjy.es.annotation.EsDocument;
import com.liuyjy.es.enums.Operator;
import com.liuyjy.es.model.req.AggregationCondition;
import com.liuyjy.es.model.req.GeoCondition;
import com.liuyjy.es.model.req.HighlightParam;
import com.liuyjy.es.model.req.QueryCondition;
import com.liuyjy.es.model.vo.*;
import com.liuyjy.es.utils.EsCacheManager;
import com.liuyjy.es.utils.EsEntityUtils;
import com.liuyjy.es.wrapper.EsLambdaQueryWrapper;
import com.liuyjy.es.wrapper.EsLambdaUpdateWrapper;
import com.liuyjy.es.wrapper.EsQueryWrapper;
import com.liuyjy.es.wrapper.EsUpdateWrapper;
import com.liuyjy.utils.utils.EmptyUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Cancellable;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.index.query.*;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.ParsedAvg;
import org.elasticsearch.search.aggregations.metrics.ParsedMax;
import org.elasticsearch.search.aggregations.metrics.ParsedMin;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

import static com.liuyjy.es.enums.MatchType.TERM;

/**
 * @Author liuyjy
 * @Date 2025/2/24 13:00
 * @Description: 核心Service实现
 **/
@Service
@RequiredArgsConstructor
@Slf4j
public class EsServiceImpl implements EsService {
    @Autowired
    private RestHighLevelClient client;

    @Autowired
    private EsCacheManager cacheManager;

    //批量操作的对象
    @Autowired
    private BulkProcessor bulkProcessor;

    /**
     * 创建索引
     * <p>
     * 此方法重写自父类，用于创建一个指定名称的索引，不包含任何设置或映射信息
     * 它通过调用另一个重载的createIndex方法来实现，传递null作为额外的参数
     *
     * @param indexName 索引的名称
     * @return 如果索引创建成功，则返回true；否则返回false
     */
    @Override
    public boolean createIndex(String indexName) {
        return createIndex(indexName, null, null);
    }

    /**
     * 创建索引
     * <p>
     * 此方法重写自父类，用于创建一个指定名称的索引，并包含额外的设置和映射信息
     * 它通过调用另一个重载的createIndex方法来实现，传递额外的设置和映射信息作为参数
     *
     * @param clazz 类
     * @throws EsException 如果索引创建过程中发生异常，则抛出此异常
     */
    @Override
    public <T> boolean createIndex(Class<T> clazz) {
        EsDocument document = clazz.getAnnotation(EsDocument.class);
        if (document == null) {
            throw new EsException("实体类缺少@EsDocument注解");
        }
        String indexName = document.indexName();
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("创建索引:索引名称不能为空");
        }
        int replicas = document.replicas();
        int shards = document.shards();
        Map<String, Object> settings = new HashMap<>();
        Map<String, Object> mapping = EsEntityUtils.buildMapping(clazz);
        settings.put("number_of_replicas", replicas);
        settings.put("number_of_shards", shards);
        return createIndex(indexName, settings, mapping);
    }

    /**
     * 创建索引
     * <p>
     * 此方法重写自父类，用于创建一个指定名称的索引，并包含额外的设置和映射信息
     * 它通过调用另一个重载的createIndex方法来实现，传递额外的设置和映射信息作为参数
     *
     * @param clazz 实体
     * @return 如果索引创建成功，则返回true；否则返回false
     */
    @Override
    public <T> boolean deleteIndex(Class<T> clazz) {
        EsDocument document = clazz.getAnnotation(EsDocument.class);
        if (document == null) {
            throw new EsException("实体类缺少@EsDocument注解");
        }
        String indexName = document.indexName();
        return deleteIndex(indexName);
    }

    /**
     * 检查索引是否存在
     *
     * @param clazz 类
     * @return 如果索引存在返回true，否则返回false
     * <p>
     * 此方法重写自父类，用于检查一个指定名称的索引是否存在
     * 它通过调用另一个重载的indexExists方法来实现，传递索引名称作为参数
     */
    @Override
    public <T> boolean indexExists(Class<T> clazz) {
        EsDocument document = clazz.getAnnotation(EsDocument.class);
        if (document == null) {
            throw new EsException("实体类缺少@EsDocument注解");
        }
        String indexName = document.indexName();
        return indexExists(indexName);
    }

    /**
     * 创建一个新的索引
     *
     * @param indexName 索引的名称
     * @param settings  索引的设置，如分片和副本的数量等
     * @param mappings  索引的映射，定义字段的类型
     * @return 如果索引创建成功返回true，否则返回false
     * <p>
     * 此方法负责向Elasticsearch集群发送创建索引的请求
     * 它允许在创建时指定索引的设置和映射，以满足特定的需求
     * 如果在创建过程中遇到IO异常，将抛出自定义的EsException异常
     */
    @Override
    public boolean createIndex(String indexName, Map<String, Object> settings, Map<String, Object> mappings) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("创建索引！索引名称不能为空");
        }
        // 创建一个创建索引请求实例，并设置索引名称
        CreateIndexRequest request = new CreateIndexRequest(indexName);

        // 如果提供了索引设置，则将其添加到请求中
        if (settings != null) {
            request.settings(settings);
        }

        // 如果提供了映射信息，则将其添加到请求中
        if (mappings != null) {
            request.mapping(mappings);
        }

        // 尝试执行创建索引请求，并处理响应
        try {
            CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
            boolean acknowledged = response.isAcknowledged();
            if (!acknowledged) {
                log.error("创建索引失败: indexNam{},response={}", indexName, response);
            } else {
                log.info("创建索引成功: " + indexName);
            }
            // 返回索引是否成功创建
            return acknowledged;
        } catch (IOException e) {
            // 如果发生IO异常，抛出自定义的EsException异常
            throw new EsException("创建索引失败", e);
        }
    }

    /**
     * 删除指定的 Elasticsearch 索引
     *
     * @param indexName 要删除的索引名称
     * @return 如果删除操作被确认，则返回 true；否则返回 false
     * @throws EsException 如果删除索引时发生错误，则抛出此异常
     */
    @Override
    public boolean deleteIndex(String indexName) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("删除索引！索引名称不能为空");
        }
        // 创建删除索引请求对象
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        try {
            // 执行删除索引操作，并获取响应结果
            AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);
            // 返回删除操作是否被确认
            return response.isAcknowledged();
        } catch (IOException e) {
            // 如果发生 IO 异常，则抛出自定义异常
            throw new EsException("删除索引失败", e);
        }
    }

    /**
     * 清空指定索引中的所有文档
     * 此方法通过删除查询来清空索引，它使用了一个匹配所有文档的查询
     * 如果操作成功，返回true；如果失败，抛出自定义异常
     *
     * @param clazz 实体类
     * @return 如果有文档被删除成功，则返回true，否则返回false
     * @throws EsException 如果清空索引失败，则抛出此异常
     */
    @Override
    public <T> boolean clearIndex(Class<T> clazz) {
        EsDocument document = clazz.getAnnotation(EsDocument.class);
        if (document == null) {
            throw new EsException("实体类缺少@EsDocument注解");
        }
        String indexName = document.indexName();

        return clearIndex(indexName);
    }

    /**
     * 清空指定索引中的所有文档
     * 此方法通过删除查询来清空索引，它使用了一个匹配所有文档的查询
     * 如果操作成功，返回true；如果失败，抛出自定义异常
     *
     * @param indexName 要清空的索引名称
     * @return 如果有文档被删除成功，则返回true，否则返回false
     * @throws EsException 如果清空索引失败，则抛出此异常
     */
    @Override
    public boolean clearIndex(String indexName) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("清空索引！索引名称不能为空");
        }
        // 创建一个删除查询请求，针对指定的索引
        DeleteByQueryRequest request = new DeleteByQueryRequest(indexName);
        // 设置查询为匹配所有文档，即删除索引中的所有文档
        request.setQuery(QueryBuilders.matchAllQuery());
        try {
            // 执行删除查询，并获取响应结果
            BulkByScrollResponse response = client.deleteByQuery(request, RequestOptions.DEFAULT);
            // 如果有文档被删除成功，返回true
            return response.getDeleted() > 0;
        } catch (IOException e) {
            // 如果操作失败，抛出自定义异常
            throw new EsException("清空索引失败", e);
        }
    }

    /**
     * 检查指定的索引是否存在于Elasticsearch中
     *
     * @param indexName 索引的名称
     * @return 如果索引存在返回true，否则返回false
     * @throws EsException 如果检查索引是否存在失败时抛出此异常
     */
    @Override
    public boolean indexExists(String indexName) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("检查指定的索引是否存在！索引名称不能为空");
        }
        // 创建一个GetIndexRequest对象，用于检查索引是否存在
        GetIndexRequest request = new GetIndexRequest(indexName);
        try {
            // 使用client的indices()方法检查索引是否存在
            return client.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            // 如果发生IOException，将其包装成EsException并抛出
            throw new EsException("检查索引是否存在失败", e);
        }
    }


    /**
     * 更新指定索引的映射
     *
     * @param indexName 索引名称
     * @param mapping   要更新的映射内容
     * @return 返回更新映射是否成功
     * <p>
     * 此方法用于向Elasticsearch索引中更新映射它首先创建一个PutMappingRequest对象，
     * 设置要更新的索引名称和映射内容，然后通过client.indices().putMapping()方法执行映射更新操作
     * 如果操作成功，返回true，否则抛出EsException异常
     */
    @Override
    public boolean updateMapping(String indexName, Map<String, Object> mapping) {
        if (EmptyUtil.isEmpty(indexName) || EmptyUtil.isEmpty(mapping)) {
            throw new EsException("更新指定索引的映射！条件为空！");
        }
        try {
            // 创建一个更新映射的请求对象
            PutMappingRequest request = new PutMappingRequest(indexName);
            // 设置映射内容
            request.source(mapping);

            // 执行映射更新操作，并返回操作是否成功
            return client.indices().putMapping(request, RequestOptions.DEFAULT).isAcknowledged();
        } catch (IOException e) {
            // 如果发生IOException，抛出自定义的EsException异常
            throw new EsException("ES-003", "Update mapping failed", e);
        }
    }

    /**
     * 更新指定索引的设置
     *
     * @param indexName 索引名称，用于指定需要更新设置的索引
     * @param settings  一个包含设置信息的映射，用于更新索引的设置
     * @return 返回更新操作是否成功如果返回true，则表示设置已成功更新
     * @throws EsException 如果更新操作失败，则抛出此自定义异常
     */
    @Override
    public boolean updateSetting(String indexName, Map<String, Object> settings) {
        if (EmptyUtil.isEmpty(indexName) || EmptyUtil.isEmpty(settings)) {
            throw new EsException("更新指定索引的设置！条件为空！");
        }
        try {
            // 创建一个更新映射的请求对象
            UpdateSettingsRequest request = new UpdateSettingsRequest(indexName);
            // 设置映射内容
            request.settings(settings);

            // 执行映射更新操作，并返回操作是否成功
            return client.indices().putSettings(request, RequestOptions.DEFAULT).isAcknowledged();
        } catch (IOException e) {
            // 如果发生IOException，抛出自定义的EsException异常
            throw new EsException("ES-003", "Update mapping failed", e);
        }
    }



    /**
     * 检查指定索引中是否存在具有指定ID的文档
     *
     * @param indexName 索引名称，对应于存储文档的容器
     * @param id        文档的唯一标识符
     * @return 如果文档存在则返回true，否则返回false
     * <p>
     * 此方法使用GetRequest对象来查询Elasticsearch客户端，以检查指定索引中是否存在具有给定ID的文档
     * 当存在IO异常时，会抛出自定义的EsException异常，以通知调用者查询失败
     */
    @Override
    public boolean exists(String indexName, String id) {
        if (EmptyUtil.isEmpty(indexName) || EmptyUtil.isEmpty(id)) {
            throw new EsException("数据是否存在！条件为空");
        }
        // 创建GetRequest对象，指定索引名称和文档ID
        GetRequest request = new GetRequest(indexName, id);
        try {
            // 使用client执行存在性检查并返回结果
            return client.exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            // 如果发生IO异常，抛出自定义异常，指示检查文档是否存在失败
            throw new EsException("检查文档是否存在失败", e);
        }
    }

    @Override
    public <T> List<T> list(EsQueryWrapper<T> wrapper) {
//        String cacheKey = buildCacheKey(wrapper);
//        List<T> cached = cacheManager.get(cacheKey);
//        if (cached != null) return cached;
        // 创建搜索请求，指定要搜索的索引
        SearchRequest request = wrapper.buildSearchRequest();
        // 执行搜索请求并处理响应
        SearchResponse response = getSearchResponse(request);
        // 执行搜索请求并处理响应
        return getList(wrapper.getEntityClass(), response);
    }


    @Override
    public <T> List<T> list(EsLambdaQueryWrapper<T> wrapper) {
        // 创建搜索请求，指定要搜索的索引
        SearchRequest request = wrapper.buildSearchRequest();

        // 执行搜索请求并处理响应
        SearchResponse response = getSearchResponse(request);
        // 执行搜索请求并处理响应
        return getList(wrapper.getEntityClass(), response);
    }

    private <T> List<T> getList(Class<T> entityClass, SearchResponse response) {

        // 将搜索结果转换为指定类型的列表
        return Arrays.stream(response.getHits().getHits())
                .map(hit -> JSON.parseObject(hit.getSourceAsString(), entityClass))
                .collect(Collectors.toList());
    }

    private SearchResponse getSearchResponse(SearchRequest request) {
        try {
            // 将搜索结果转换为指定类型的列表
            return client.search(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            // 如果发生IO异常，抛出自定义的EsException异常
            throw new EsException("查询列表失败", e);
        }
    }

    @Override
    public <T> PageResult<T> page(EsQueryWrapper<T> wrapper) {
        // 创建搜索请求，指定要搜索的索引
        SearchRequest request = wrapper.buildSearchRequest();
        // 执行搜索请求并处理响应
        SearchResponse response = getSearchResponse(request);

        // 格式化
        List<T> list = getList(wrapper.getEntityClass(), response);

        return PageResult.of(response.getHits().getTotalHits().value, wrapper.current, wrapper.size, list);

    }


    @Override
    public <T> PageResult<T> page(EsLambdaQueryWrapper<T> wrapper) {
        // 创建搜索请求，指定要搜索的索引
        SearchRequest request = wrapper.buildSearchRequest();
        // 执行搜索请求并处理响应
        SearchResponse response = getSearchResponse(request);

        // 格式化
        List<T> list = getList(wrapper.getEntityClass(), response);

        return PageResult.of(response.getHits().getTotalHits().value, wrapper.current, wrapper.size, list);
    }

    @Override
    public <T> Map<String, Object> aggregate(EsQueryWrapper<T> wrapper) {
        // 创建搜索请求，指定要搜索的索引
        SearchRequest request = wrapper.buildSearchRequest();

        // 执行搜索请求并处理响应
        SearchResponse response = getSearchResponse(request);
        // 执行搜索请求并处理响应
        return parseAggregationResponse(response.getAggregations());
    }

    @Override
    public <T> Map<String, Object> aggregate(EsLambdaQueryWrapper<T> wrapper) {
        // 创建搜索请求，指定要搜索的索引
        SearchRequest request = wrapper.buildSearchRequest();

        // 执行搜索请求并处理响应
        SearchResponse response = getSearchResponse(request);
        // 执行搜索请求并处理响应
        return parseAggregationResponse(response.getAggregations());
    }



    /**
     * 将指定文档保存到指定的Elasticsearch索引中
     * 如果文档成功创建，则返回true；否则返回false
     *
     * @param indexName 索引名称，用于指定文档应存储的位置
     * @param id        文档的唯一标识符，用于在索引中区分不同的文档
     * @param entity    要保存的文档对象，可以是任意类型，但需序列化为JSON格式
     * @return 如果文档成功保存则返回true，否则返回false
     * @throws EsException 如果保存文档过程中发生错误，则抛出此异常
     */
    @Override
    public <T> boolean save(String indexName, String id, T entity) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("文档保存:索引名称不能为空");
        }
        if (EmptyUtil.isEmpty(entity)) {
            throw new EsException("文档保存:文档不能为空");
        }
        if (EmptyUtil.isEmpty(id)) {
            throw new EsException("文档保存:文档id不能为空");
        }
        // 创建一个IndexRequest对象，用于执行文档索引操作
        IndexRequest request = new IndexRequest(indexName).id(id)
                .source(JSON.toJSONString(entity), XContentType.JSON);
        try {
            // 执行索引操作并获取响应结果
            IndexResponse response = client.index(request, RequestOptions.DEFAULT);
            //client.updateAsync(new UpdateRequest(), RequestOptions.DEFAULT, null);
            // 检查响应状态，判断文档是否成功创建
            log.info("文档创建状态：{}", response.status());
            return response.status() == RestStatus.CREATED;
        } catch (IOException e) {
            // 如果发生IO错误，抛出自定义的EsException异常
            throw new EsException("保存文档失败", e);
        }
    }

    @Override
    public <T> boolean save(T entity) {
        String indexName = EsEntityUtils.getIndexName(entity.getClass());
        String entityId = EsEntityUtils.getEntityId(entity);
        return save(indexName, entityId, entity);
    }


    @Override
    public <T> boolean save(String indexName, T entity) {
        String entityId = EsEntityUtils.getEntityId(entity);
        return save(indexName, entityId, entity);
    }

    /**
     * 将指定文档保存到指定的Elasticsearch索引中
     * 如果文档成功创建，则返回true；否则返回false
     *
     * @param indexName 索引名称，用于指定文档应存储的位置
     * @param id        文档的唯一标识符，用于在索引中区分不同的文档
     * @param entity    要保存的文档对象，可以是任意类型，但需序列化为JSON格式
     * @return 如果文档成功保存则返回true，否则返回false
     * @throws EsException 如果保存文档过程中发生错误，则抛出此异常
     */
    @Override
    public <T> Cancellable saveAsync(String indexName, String id, T entity) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("文档保存:索引名称不能为空");
        }
        if (EmptyUtil.isEmpty(entity)) {
            throw new EsException("文档保存:文档不能为空");
        }
        if (EmptyUtil.isEmpty(id)) {
            throw new EsException("文档保存:文档id不能为空");
        }
        // 创建一个IndexRequest对象，用于执行文档索引操作
        IndexRequest request = new IndexRequest(indexName).id(id)
                .source(JSON.toJSONString(entity), XContentType.JSON);
        Cancellable cancellable = client.indexAsync(request, RequestOptions.DEFAULT,
                new ActionListener<IndexResponse>() {
                    @Override
                    public void onResponse(IndexResponse indexResponse) {
                        ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                        if (shardInfo.getFailed() > 0) {
                            for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                                log.error("将id为：{}的数据存入ES时存在失败的分片，原因为：{}", request.id(), failure.getCause());
                            }
                        }
                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("{}:存储es时异常，数据信息为", request.id(), e);
                    }
                });
        return cancellable;
    }

    @Override
    public <T> Cancellable saveAsync(T entity) {
        String indexName = EsEntityUtils.getIndexName(entity.getClass());
        String entityId = EsEntityUtils.getEntityId(entity);
        return saveAsync(indexName, entityId, entity);
    }


    @Override
    public <T> Cancellable saveAsync(String indexName, T entity) {
        String entityId = EsEntityUtils.getEntityId(entity);
        return saveAsync(indexName, entityId, entity);
    }

    /**
     * 批量保存文档到指定索引
     *
     * @param indexName 索引名称，用于指定文档存储的位置
     * @param documents 文档列表，包含多个待保存的文档对象
     * @return 保存结果，如果所有文档都成功保存则返回true，否则返回false
     * <p>
     * 此方法使用Elasticsearch的Bulk API来批量索引文档，以提高索引效率
     * 它通过构造一个BulkRequest，为每个文档添加一个IndexRequest，然后执行批量操作
     * 如果在执行过程中遇到IOException，将抛出自定义的EsException异常
     */
    public <T> boolean bulkSave(String indexName, List<T> documents) {
        // 创建BulkRequest对象，用于批量执行索引、删除等操作
        BulkRequest request = new BulkRequest();
        // 遍历文档列表，为每个文档创建IndexRequest并添加到BulkRequest中

        for (T document : documents) {
            request.add(new IndexRequest(indexName)
                    .id(EsEntityUtils.getEntityId(document))
                    .source(JSON.toJSONString(document), XContentType.JSON));
        }
        // 尝试执行BulkRequest，如果成功则返回没有失败操作的结果
        try {
            BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);
            return !response.hasFailures();
        } catch (IOException e) {
            // 如果执行过程中抛出IOException，抛出自定义异常
            throw new EsException("批量保存文档失败", e);
        }
    }


    @Override
    public <T> void saveBatch(String indexName, Collection<T> entities) {
        if (EmptyUtil.isEmpty(entities) || EmptyUtil.isEmpty(indexName)) {
            throw new EsException("批量保存文档失败；条件为空！");
        }
        List<IndexRequest> indexRequests = new ArrayList<>();

        entities.forEach(e -> {
            String id = EsEntityUtils.getEntityId(e);
            setIndexRequests(e, id, indexName, indexRequests);
        });
        indexRequests.forEach(bulkProcessor::add);
    }


    @Override
    public <T> void saveBatch(Collection<T> entities) {
        if (EmptyUtil.isEmpty(entities)) {
            throw new EsException("批量保存文档失败；条件为空！");
        }
        List<IndexRequest> indexRequests = new ArrayList<>();
        for (T entity : entities) {
            String indexName = EsEntityUtils.getIndexName(entity.getClass());
            String id = EsEntityUtils.getEntityId(entity);
            setIndexRequests(entity, id, indexName, indexRequests);
        }
        indexRequests.forEach(bulkProcessor::add);
    }

    /**
     * 设置indexRequests
     *
     * @param entity        数据
     * @param id            id
     * @param indexName     索引
     * @param indexRequests indexRequests
     */
    private <T> void setIndexRequests(T entity, String id, String indexName, List<IndexRequest> indexRequests) {
        if (EmptyUtil.isEmpty(id)) {
            log.error("批量保存文档失败；id为空！");
            return;
        }
        if (EmptyUtil.isEmpty(indexName)) {
            log.error("批量保存文档失败；indexName为空！");
            return;
        }
        IndexRequest request = new IndexRequest(indexName);
        //填充id
        request.id(id);
        //先不修改id
        request.source(JSON.toJSONString(entity), XContentType.JSON);
//        OpType.CREATE：当存在相同的_id时，插入会出现异常；
//        OpType.INDEX：当存在相同_id时，插入会进行覆盖；
        //request.opType(DocWriteRequest.OpType.CREATE);
        indexRequests.add(request);
    }


    @Override
    public <T> boolean updateById(String indexName, String id, T entity) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("更新文档失败:索引名称不能为空");
        }
        if (EmptyUtil.isEmpty(entity)) {
            throw new EsException("更新文档失败:文档不能为空");
        }
        if (EmptyUtil.isEmpty(id)) {
            throw new EsException("更新文档失败:文档id不能为空");
        }
        UpdateRequest updateRequest = new UpdateRequest(indexName, id);
        try {
            UpdateResponse response = client.update(updateRequest, RequestOptions.DEFAULT);
            return response.getResult() == DocWriteResponse.Result.UPDATED;
        } catch (IOException e) {
            throw new EsException("保存文档失败", e);
        }
    }

    @Override
    public <T> boolean updateById(T entity) {
        String indexName = EsEntityUtils.getIndexName(entity.getClass());
        String id = EsEntityUtils.getEntityId(entity);
        return updateById(indexName, id, entity);
    }

    @Override
    public <T> Cancellable updateByIdAsync(String indexName, String id, T entity) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("更新文档失败:索引名称不能为空");
        }
        if (EmptyUtil.isEmpty(entity)) {
            throw new EsException("更新文档失败:文档不能为空");
        }
        if (EmptyUtil.isEmpty(id)) {
            throw new EsException("更新文档失败:文档id不能为空");
        }
        UpdateRequest updateRequest = new UpdateRequest(indexName, id);

        Cancellable cancellable = client.updateAsync(updateRequest, RequestOptions.DEFAULT,
                new ActionListener<UpdateResponse>() {
                    @Override
                    public void onResponse(UpdateResponse indexResponse) {
                        ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                        if (shardInfo.getFailed() > 0) {
                            for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                                log.error("将id为：{}的数据存入ES时存在失败的分片，原因为：{}", updateRequest.id(), failure.getCause());
                            }
                        }
                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("{}:存储es时异常，数据信息为", updateRequest.id(), e);
                    }
                });
        return cancellable;
    }

    @Override
    public <T> Cancellable updateByIdAsync(T entity) {
        String indexName = EsEntityUtils.getIndexName(entity.getClass());
        String id = EsEntityUtils.getEntityId(entity);
        return updateByIdAsync(indexName, id, entity);
    }

    @Override
    public <T> void updateBatch(String indexName, Collection<T> entities) {
        if (EmptyUtil.isEmpty(entities)) {
            throw new EsException("批量保存文档失败；条件为空！");
        }
        List<UpdateRequest> updateRequests = new ArrayList<>();
        for (T entity : entities) {

            String id = EsEntityUtils.getEntityId(entity);
            setUpdateRequests(entity, id, indexName, updateRequests);
        }
        updateRequests.forEach(bulkProcessor::add);
    }

    @Override
    public <T> void updateBatch(Collection<T> entities) {
        if (EmptyUtil.isEmpty(entities)) {
            throw new EsException("批量保存文档失败；条件为空！");
        }
        List<UpdateRequest> updateRequests = new ArrayList<>();
        for (T entity : entities) {
            String indexName = EsEntityUtils.getIndexName(entity.getClass());
            String id = EsEntityUtils.getEntityId(entity);
            setUpdateRequests(entity, id, indexName, updateRequests);
        }
        updateRequests.forEach(bulkProcessor::add);
    }

    /**
     * 设置indexRequests
     *
     * @param entity         数据
     * @param id             id
     * @param indexName      索引
     * @param updateRequests updateRequests
     */
    private <T> void setUpdateRequests(T entity, String id, String indexName, List<UpdateRequest> updateRequests) {
        if (EmptyUtil.isEmpty(id)) {
            log.error("批量修改文档失败；id为空！");
            return;
        }
        if (EmptyUtil.isEmpty(indexName)) {
            log.error("批量修改文档失败；indexName为空！");
            return;
        }
        UpdateRequest request = new UpdateRequest();
        request.index(indexName);
        //更新的id
        request.id(id);
        request.doc(JSON.toJSONString(entity), XContentType.JSON);
        updateRequests.add(request);
    }

    @Override
    public <T> long updateByQuery(EsUpdateWrapper<T> wrapper) {
        return updateByQuery(wrapper.buildUpdateRequest());
    }

    @Override
    public <T> long updateByQuery(EsLambdaUpdateWrapper<T> wrapper) {
        return updateByQuery(wrapper.buildUpdateRequest());
    }

    private long updateByQuery(UpdateByQueryRequest request) {
        try {
            BulkByScrollResponse response = client.updateByQuery(request, RequestOptions.DEFAULT);
            //cacheManager.clear(); // 更新后清空缓存
            return response.getUpdated();
        } catch (IOException e) {
            throw new EsException("条件更新失败", e);
        }
    }

    @Override
    public <T> Cancellable updateByQueryAsync(EsUpdateWrapper<T> wrapper) {
        return updateByQueryAsync(wrapper.buildUpdateRequest());
    }

    @Override
    public <T> Cancellable updateByQueryAsync(EsLambdaUpdateWrapper<T> wrapper) {
        return updateByQueryAsync(wrapper.buildUpdateRequest());
    }

    private Cancellable updateByQueryAsync(UpdateByQueryRequest request) {
        return client.updateByQueryAsync(request, RequestOptions.DEFAULT,
                new ActionListener<BulkByScrollResponse>() {
                    @Override
                    public void onResponse(BulkByScrollResponse bulkByScrollResponse) {

                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("存储es时异常，数据信息为", e);
                    }

                });
    }

    @Override
    public boolean deleteById(String indexName, String id) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("删除文档失败；索引名称为空！");
        }
        if (EmptyUtil.isEmpty(id)) {
            throw new EsException("删除文档失败；id为空！");
        }
        DeleteRequest deleteRequest = new DeleteRequest();
        deleteRequest.index(indexName);
        deleteRequest.id(id);
        try {
            DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
            log.info("删除文档结果；Result为{}！", deleteResponse.getResult());
            return deleteResponse.getResult() == DocWriteResponse.Result.DELETED;
        } catch (IOException e) {
            throw new EsException("删除文档结果", e);
        }
    }

    @Override
    public Cancellable deleteByIdAsync(String indexName, String id) {
        if (EmptyUtil.isEmpty(indexName)) {
            throw new EsException("删除文档失败；索引名称为空！");
        }
        if (EmptyUtil.isEmpty(id)) {
            throw new EsException("删除文档失败；id为空！");
        }
        DeleteRequest deleteRequest = new DeleteRequest();
        deleteRequest.index(indexName);
        deleteRequest.id(id);
        return client.deleteAsync(deleteRequest, RequestOptions.DEFAULT,
                new ActionListener<DeleteResponse>() {
                    @Override
                    public void onResponse(DeleteResponse deleteResponse) {

                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("删除文档异常", e);
                    }

                });
    }

    @Override
    public <T> long deleteByQuery(EsUpdateWrapper<T> wrapper) {
        return deleteByQuery(wrapper.buildDeleteRequest());
    }

    @Override
    public <T> long deleteByQuery(EsLambdaUpdateWrapper<T> wrapper) {
        return deleteByQuery(wrapper.buildDeleteRequest());
    }

    private long deleteByQuery(DeleteByQueryRequest request) {
        try {
            BulkByScrollResponse response = client.deleteByQuery(request, RequestOptions.DEFAULT);
            //cacheManager.clear(); // 更新后清空缓存
            return response.getUpdated();
        } catch (IOException e) {
            throw new EsException("条件更新失败", e);
        }
    }

    @Override
    public <T> Cancellable deleteByQueryAsync(EsUpdateWrapper<T> wrapper) {
        return deleteByQueryAsync(wrapper.buildDeleteRequest());
    }

    @Override
    public <T> Cancellable deleteByQueryAsync(EsLambdaUpdateWrapper<T> wrapper) {
        return deleteByQueryAsync(wrapper.buildDeleteRequest());
    }

    private Cancellable deleteByQueryAsync(DeleteByQueryRequest request) {
        return client.deleteByQueryAsync(request, RequestOptions.DEFAULT,
                new ActionListener<BulkByScrollResponse>() {
                    @Override
                    public void onResponse(BulkByScrollResponse bulkByScrollResponse) {

                    }

                    @Override
                    public void onFailure(Exception e) {
                        log.error("存储es时异常，数据信息为", e);
                    }

                });
    }

    /**
     * 根据ID查询指定索引中的文档，并将其解析为指定类型的对象
     *
     * @param indexName 索引名称，即文档所属的索引
     * @param id        文档的ID
     * @param clazz     要解析的文档类型，用于指定返回对象的类型
     * @param <T>       泛型参数，表示返回对象的类型
     * @return 解析后的对象，如果文档不存在则返回null
     * @throws EsException 如果查询过程中发生错误，则抛出此异常
     */
    @Override
    public <T> T getById(String indexName, String id, Class<T> clazz) {
        // 创建获取文档的请求
        GetRequest request = new GetRequest(indexName, id);
        try {
            // 执行获取文档的请求
            GetResponse response = client.get(request, RequestOptions.DEFAULT);
            // 检查文档是否存在
            if (response.isExists()) {
                // 文档存在时，将其内容解析为指定类型的对象并返回
                return JSON.parseObject(response.getSourceAsString(), clazz);
            }
            // 文档不存在时，返回null
            return null;
        } catch (IOException e) {
            // 查询过程中发生错误时，抛出自定义异常
            throw new EsException("根据ID查询文档失败", e);
        }
    }

    /**
     * 根据指定的索引和查询条件搜索文档列表
     *
     * @param indices   要搜索的索引数组
     * @param condition 查询条件对象
     * @param clazz     要转换的类类型
     * @return 返回转换后的对象列表
     * <p>
     * 此方法使用Elasticsearch客户端执行搜索操作，并将搜索结果转换为指定类型的列表
     * 它首先构建一个搜索请求，然后根据提供的查询条件构建查询源，最后发送请求并处理响应
     */
    @Override
    public <T> List<T> searchList(String[] indices, QueryCondition condition, Class<T> clazz) {
        SearchResponse response = getSearchResponse(indices, condition);

        // 执行搜索请求并处理响应
        return getList(clazz, response);
    }

    private <T> SearchResponse getSearchResponse(String[] indices, QueryCondition condition) {
        // 创建搜索请求，指定要搜索的索引
        SearchRequest request = new SearchRequest(indices);

        // 构建查询源，包括查询条件、过滤器、排序等
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        // 根据提供的查询条件构建查询源
        buildQuery(sourceBuilder, condition);

        // 将构建好的查询源设置到请求中
        request.source(sourceBuilder);

        // 执行搜索请求并处理响应
        return getSearchResponse(request);
    }


    /**
     * 在指定索引中执行分页搜索
     *
     * @param indices   搜索的索引数组
     * @param condition 查询条件对象，包含查询的具体参数
     * @param clazz     需要返回的对象类型，用于将搜索结果转换为指定的Java对象类型
     * @return 返回一个包含分页结果的PageResult对象，包括总命中数和数据列表
     * <p>
     * 此方法负责构建搜索请求，执行搜索操作，并将搜索结果解析为指定类型的分页结果对象
     * 它使用了Elasticsearch提供的SearchRequest和SearchResponse类来处理搜索请求和响应
     */
    @Override
    public <T> PageResult<T> searchPage(String[] indices, QueryCondition condition, Class<T> clazz) {
        PageParam pageParam = condition.getPageParam();
        if (pageParam == null) {
            pageParam = new PageParam();
        }

        SearchResponse response = getSearchResponse(indices, condition);
        // 格式化
        List<T> list = getList(clazz, response);

        return PageResult.of(response.getHits().getTotalHits().value, pageParam.getPageNum(), pageParam.getPageNum(), list);
    }


    /**
     * 根据地理位置条件查询指定索引中的数据
     *
     * @param indexName 索引名称，用于指定要查询的索引
     * @param condition 地理位置查询条件，包含查询所需的各种参数
     * @param clazz     查询结果要转换成的对象类型
     * @return 返回查询到的结果列表，列表中的元素类型为clazz指定的类型
     * <p>
     * 此方法主要用于执行基于地理位置的搜索操作，它根据提供的地理位置条件在指定索引中查找数据，
     * 并将查询到的结果转换为指定类型的对象列表返回
     */
    @Override
    public <T> List<T> geoSearch(String indexName, GeoCondition condition, Class<T> clazz) {
        // 创建搜索请求对象，指定要查询的索引
        SearchRequest request = new SearchRequest(indexName);
        // 创建搜索源构建器，用于构建查询的各种条件
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        // 构建地理位置查询
        GeoDistanceQueryBuilder geoQuery = QueryBuilders.geoDistanceQuery(condition.getField())
                .point(condition.getCenterPoint().getLat(), condition.getCenterPoint().getLon())
                .distance(condition.getDistance());
        sourceBuilder.query(geoQuery);

        // 添加排序
        if (condition.getSortOrder() != null) {
            sourceBuilder.sort(SortBuilders.geoDistanceSort(condition.getField(),
                            condition.getCenterPoint().getLat(), condition.getCenterPoint().getLon())
                    .order(condition.getSortOrder()));
        }

        request.source(sourceBuilder);
        try {
            // 执行搜索操作
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 解析搜索结果，将其转换为指定类型的对象列表
            return Arrays.stream(response.getHits().getHits())
                    .map(hit -> JSON.parseObject(hit.getSourceAsString(), clazz))
                    .collect(Collectors.toList());
        } catch (IOException e) {
            // 如果查询过程中发生IO异常，则抛出自定义的EsException异常
            throw new EsException("地理位置查询失败", e);
        }
    }

    /**
     * 根据指定的索引名称和聚合条件执行聚合查询
     *
     * @param indexName 索引名称，用于指定查询的索引
     * @param condition 聚合条件对象，包含具体的聚合操作及其参数
     * @return 返回包含聚合结果的映射，其中键为聚合名称，值为聚合结果
     * @throws EsException 如果聚合类型不支持或查询过程中出现IO异常，则抛出此异常
     */
    @Override
    public Map<String, Object> aggregate(String indexName, AggregationCondition condition) {
        // 创建搜索请求对象，指定索引名称
        SearchRequest request = new SearchRequest(indexName);
        // 创建搜索源构建器，用于配置查询的各种参数
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        // 构建聚合条件
        condition.getAggregations().forEach((name, agg) -> {
            // 根据聚合类型执行不同的聚合操作
            switch (agg.getType()) {
                case TERMS:
                    // 执行terms聚合，通常用于分组
                    sourceBuilder.aggregation(AggregationBuilders.terms(name).field(agg.getField()));
                    break;
                case AVG:
                    // 执行平均值聚合
                    sourceBuilder.aggregation(AggregationBuilders.avg(name).field(agg.getField()));
                    break;
                case MAX:
                    // 执行最大值聚合
                    sourceBuilder.aggregation(AggregationBuilders.max(name).field(agg.getField()));
                    break;
                case MIN:
                    // 执行最小值聚合
                    sourceBuilder.aggregation(AggregationBuilders.min(name).field(agg.getField()));
                    break;
                default:
                    // 如果聚合类型不支持，则抛出异常
                    throw new EsException("不支持的聚合类型");
            }
        });

        // 将构建好的搜索源配置到请求对象中
        request.source(sourceBuilder);
        // 执行搜索请求并处理响应
        try {
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 解析聚合响应并返回结果
            return parseAggregationResponse(response.getAggregations());
        } catch (IOException e) {
            // 如果查询过程中出现IO异常，抛出自定义异常
            throw new EsException("聚合查询失败", e);
        }
    }

    private Map<String, Object> parseAggregationResponse(Aggregations aggregations) {
        Map<String, Object> result = new HashMap<>();
        aggregations.forEach(aggregation -> {
            if (aggregation instanceof ParsedTerms) {
                List<? extends Terms.Bucket> buckets = ((ParsedTerms) aggregation).getBuckets();
                result.put(aggregation.getName(), buckets.stream()
                        .collect(Collectors.toMap(Terms.Bucket::getKeyAsString, Terms.Bucket::getDocCount)));
            } else if (aggregation instanceof ParsedAvg) {
                result.put(aggregation.getName(), ((ParsedAvg) aggregation).getValue());
            } else if (aggregation instanceof ParsedMax) {
                result.put(aggregation.getName(), ((ParsedMax) aggregation).getValue());
            } else if (aggregation instanceof ParsedMin) {
                result.put(aggregation.getName(), ((ParsedMin) aggregation).getValue());
            }
        });
        return result;
    }


    /**
     * 构建查询
     * 根据给定的查询条件构建Elasticsearch的查询DSL
     *
     * @param sourceBuilder 查询构建器，用于构建查询DSL
     * @param condition     查询条件，包含查询的各种条件和参数
     */
    private void buildQuery(SearchSourceBuilder sourceBuilder, QueryCondition condition) {
        // 创建一个布尔查询构建器
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

        // 遍历必须满足的条件，添加到布尔查询的must子句中
        condition.getMustConditions().forEach(mc -> boolQuery.must(buildQuery(mc)));

        // 遍历可能满足的条件，添加到布尔查询的should子句中
        condition.getShouldConditions().forEach(sc -> boolQuery.should(buildQuery(sc)));

        // 遍历操作条件，添加到布尔查询的should子句中
        List<OperatorCondition> operatorConditionList = condition.getOperatorConditions();
        buildBoolQuery(boolQuery, operatorConditionList);


        // 遍历范围条件，添加到布尔查询的filter子句中
        condition.getRangeConditions().forEach(rc -> boolQuery.filter(QueryBuilders.rangeQuery(rc.getField())
                .from(rc.getFrom()).to(rc.getTo())));

        // 将构建好的布尔查询添加到查询源构建器中
        sourceBuilder.query(boolQuery);

        // 分页
        // 设置查询的起始位置和查询条数，实现分页查询
        sourceBuilder.from(condition.getPageParam().getOffset());
        sourceBuilder.size(condition.getPageParam().getPageSize());

        // 排序
        // 如果排序参数不为空，则遍历排序字段，添加排序条件
        List<SortField> sortFields = condition.getSortFields();
        //判断集合是否为空
        if (null != sortFields && !sortFields.isEmpty()) {
            sortFields.forEach(sf -> {
                // 如果是地理位置排序，则添加地理位置排序条件
                if (sf.isGeoSort()) {
                    sourceBuilder.sort(SortBuilders.geoDistanceSort(sf.getField(),
                                    condition.getGeoCondition().getCenterPoint().getLat(),
                                    condition.getGeoCondition().getCenterPoint().getLon())
                            .order(sf.getOrder()));
                } else {
                    // 否则，添加普通字段排序条件
                    sourceBuilder.sort(sf.getField(), sf.getOrder());
                }
            });
        }
    }


    /**
     * 根据匹配条件构建查询构建器
     * 此方法旨在根据提供的匹配条件对象中的信息来构建不同类型的查询
     * 它使用了Elasticsearch的QueryBuilders工具类来构造查询对象
     *
     * @param condition 匹配条件对象，包含查询类型、字段和值
     * @return 返回根据匹配条件构建的查询构建器对象
     * @throws EsException 如果匹配条件中的查询类型不受支持，则抛出此异常
     */
    private QueryBuilder buildQuery(MatchCondition condition) {
        // 根据匹配条件的类型选择相应的查询构建方法
        switch (condition.getType()) {
            case TERM:
                return QueryBuilders.termQuery(condition.getField(), condition.getValue());
            case MATCH:
                return QueryBuilders.matchQuery(condition.getField(), condition.getValue())
                        .analyzer("ik_smart"); // 指定搜索分词器
            case MATCH_PHRASE:
                return QueryBuilders.matchPhraseQuery(condition.getField(), condition.getValue());
            case TERMS:
                return QueryBuilders.termsQuery(condition.getField(), (Collection<?>) condition.getValue());
            case PREFIX:
                return QueryBuilders.prefixQuery(condition.getField(), condition.getValue().toString());
            case WILDCARD:
                return QueryBuilders.wildcardQuery(condition.getField(), "*" + condition.getValue() + "*");
            case FUZZY:
                return QueryBuilders.fuzzyQuery(condition.getField(), condition.getValue());
            default:
                throw new UnsupportedOperationException("不支持的查询类型: " + condition.getType());
        }
    }

    private void handleSingleValueRange(BoolQueryBuilder boolQuery, OperatorCondition rc) {
        String field = rc.getField();
        Object value = rc.getValue();
        Operator operator = rc.getOperator();
        switch (operator) {
            case GT:
                boolQuery.filter(QueryBuilders.rangeQuery(field).gt(value));
                break;
            case GTE:
                boolQuery.filter(QueryBuilders.rangeQuery(field).gte(value));
                break;
            case LT:
                boolQuery.filter(QueryBuilders.rangeQuery(field).lt(value));
                break;
            case LTE:
                boolQuery.filter(QueryBuilders.rangeQuery(field).lte(value));
                break;
            case EQ:
                boolQuery.filter(QueryBuilders.termQuery(field, value));
                break;
            case NEQ:
                boolQuery.mustNot(QueryBuilders.termQuery(field, value));
                break;
        }
    }

    private void buildBoolQuery(BoolQueryBuilder boolQuery, List<OperatorCondition> list) {
        if (null == list || list.isEmpty()) {
            return;
        }
        for (OperatorCondition condition : list) {
            handleSingleValueRange(boolQuery, condition);
        }
    }


    private void handleRangeQuery(BoolQueryBuilder boolQuery, RangeCondition rc) {
        RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery(rc.getField());
        if (rc.getFrom() != null) {
            rangeQuery.from(rc.getFrom(), rc.isIncludeLower());
        }
        if (rc.getTo() != null) {
            rangeQuery.to(rc.getTo(), rc.isIncludeUpper());
        }
        boolQuery.filter(rangeQuery);
    }

    /**
     * 高亮查询
     *
     * @param sourceBuilder
     * @param highlight
     */
    private void buildHighlight(SearchSourceBuilder sourceBuilder, HighlightParam highlight) {
        if (highlight != null && highlight.getFields() != null) {
            HighlightBuilder highlightBuilder = new HighlightBuilder();
            highlight.getFields().forEach(field -> highlightBuilder.field(field)
                    .preTags(highlight.getPreTags())
                    .postTags(highlight.getPostTags()));
            sourceBuilder.highlighter(highlightBuilder);
        }
    }

    //    public <T> List<T> executeSql(String sql, Class<T> clazz) {
//        SqlTranslateRequest request = new SqlTranslateRequest();
//        request.query(sql);
//        try {
//            SqlTranslateResponse response = client.sql().translate(request, RequestOptions.DEFAULT);
//            SearchSourceBuilder builder = SearchSourceBuilder.fromXContent(
//                    XContentFactory.xContent(response.source()));
//            return parseSearchResult(client.search(new SearchRequest().source(builder), RequestOptions.DEFAULT), clazz);
//        } catch (IOException e) {
//            throw new EsQueryException("SQL 执行失败", e);
//        }
//    }


}
